package com.sojson.util.mq.rabbit.immit.interfaces.impl;

import java.io.IOException;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitOperations;

import com.sojson.util.StringUtil;
import com.sojson.util.mq.rabbit.enums.ERabbitExchange;

/**
 * rabbitmq_delayed_message_exchange插件</BR>
 * 用来让消息延迟一段时间再发送到交换机,以实现延迟发送消息</BR>
 * 创建交换机的时候需要在参数里面指定交换机的类型{@link com.sojson.util.mq.rabbit.enums.ERabbitExchange}</BR>
 * 发送的消息如果要实现延迟发送,需要在发送的参数里面指定延迟时间(单位:毫秒)</BR>
 * 
 * @author liu
 * @date 2022-06-16
 */
public class RabbitDelayMsgUtilImpl {

    private static RabbitDelayMsgUtilImpl instances;
    /** {@link com.sojson.util.mq.rabbit.code.interfaces.impl.RabbitDirectUtilImpl}的扩展类 */
    private static RabbitDirectDelayMsgUtilImpl direct;
    /** {@link com.sojson.util.mq.rabbit.code.interfaces.impl.RabbitTopicUtilImpl}的扩展类 */
    private static RabbitTopicDelayMsgUtilImpl topic;
    /** {@link com.sojson.util.mq.rabbit.code.interfaces.impl.RabbitFanoutUtilImpl}的扩展类 */
    private static RabbitFanoutDelayMsgUtilImpl fanout;
    /** {@link com.sojson.util.mq.rabbit.code.interfaces.impl.RabbitHeadersUtilImpl}的扩展类 */
    private static RabbitHeadersDelayMsgUtilImpl headers;

    /** 获取 {@link com.sojson.util.mq.rabbit.code.interfaces.impl.RabbitDirectUtilImpl}的扩展类 */
    public RabbitDirectDelayMsgUtilImpl getDirect() {
        if (isBlankObject(direct)) {
            direct = new RabbitDelayMsgUtilImpl().new RabbitDirectDelayMsgUtilImpl();
        }
        return direct;
    }

    /** 获取 {@link com.sojson.util.mq.rabbit.code.interfaces.impl.RabbitTopicUtilImpl}的扩展类 */
    public RabbitTopicDelayMsgUtilImpl getTopic() {
        if (isBlankObject(topic)) {
            topic = new RabbitDelayMsgUtilImpl().new RabbitTopicDelayMsgUtilImpl();
        }
        return topic;
    }

    /** 获取 {@link com.sojson.util.mq.rabbit.code.interfaces.impl.RabbitFanoutUtilImpl}的扩展类 */
    public RabbitFanoutDelayMsgUtilImpl getFanout() {
        if (isBlankObject(fanout)) {
            fanout = new RabbitDelayMsgUtilImpl().new RabbitFanoutDelayMsgUtilImpl();
        }
        return fanout;
    }

    /** 获取{@link com.sojson.util.mq.rabbit.code.interfaces.impl.RabbitHeadersUtilImpl}的扩展类 */
    public RabbitHeadersDelayMsgUtilImpl getHeaders() {
        if (isBlankObject(headers)) {
            headers = new RabbitDelayMsgUtilImpl().new RabbitHeadersDelayMsgUtilImpl();
        }
        return headers;
    }

    /**
     * {@link com.sojson.util.mq.rabbit.code.interfaces.impl.RabbitDirectUtilImpl}的扩展类
     * 
     * @author liu
     * @date 2022-06-17
     */
    public class RabbitDirectDelayMsgUtilImpl extends RabbitDirectUtilImpl {

        /**
         * 开始生产消息
         * 
         * @param channel
         * @param exchange      交换机的名称
         * @param routingKey    消息的Key(如果没有交换机,此时是将消息发布到的队列名)
         * @param body          内容
         * @param delayTime     发送延迟时间
         * @return
         * @throws IOException
         */
        public void basicPublish(RabbitOperations channel, String exchange, String routingKey, String body,
            Long delayTime) throws IOException {
            basicPublish(channel, exchange, routingKey, false, false, body, delayTime);
        }

        /**
         * 开始生产消息
         * 
         * @param channel
         * @param exchange      交换机的名称
         * @param routingKey    消息的Key(如果没有交换机,此时是将消息发布到的队列名)
         * @param mandatory     是否开启消息故障检测模式
         *                      它只会让RabbitMq向你通知失败,而不会通知成功
         *                      设为true的时候,使用前要先调用channel.addReturnListener(new ReturnCallback(){})
         *                      channel.addReturnListener(new ReturnListener(){})为异步返回
         *                      如果为true,消息不能路由到指定的队列时,会触发channel.BasicReturn事件
         *                      如果为false,则broker会直接将消息丢弃
         * @param immediate     RabbitMQ 3.0版本开始去掉了对immediate参数的支持
         *                      原因是因为immediate参数会影响镜像队列的性能,增加了代码的复杂度,建议采用TTL和DLX的方法替代
         *                      当immediate参数设置为true时,如果交换器再将消息路由到一个队列中,否则将消息返回给生产者
         *                      immediate参数告诉服务器,如果该消息关联的队列上有消费者,则立刻投递
         *                      如果所有匹配的队列上都没有消费者,则直接将消息返还给生产者,不用将消息存于的队列而等待消费者
         * @param body          内容
         * @param delayTime     发送延迟时间
         * @return
         * @throws IOException
         */
        public void basicPublish(RabbitOperations channel, String exchange, String routingKey, boolean mandatory,
            boolean immediate, String body, Long delayTime) throws IOException {
            basicPublish(channel, exchange, routingKey, mandatory, immediate, getBasicProperties(), body, delayTime);
        }

        /**
         * 开始生产消息
         * 
         * @param channel
         * @param exchange      交换机的名称
         * @param routingKey    消息的Key(如果没有交换机,此时是将消息发布到的队列名)
         * @param mandatory     是否开启消息故障检测模式
         *                      它只会让RabbitMq向你通知失败,而不会通知成功
         *                      设为true的时候,使用前要先调用channel.addReturnListener(new ReturnCallback(){})
         *                      channel.addReturnListener(new ReturnListener(){})为异步返回
         *                      如果为true,消息不能路由到指定的队列时,会触发channel.BasicReturn事件
         *                      如果为false,则broker会直接将消息丢弃
         * @param immediate     RabbitMQ 3.0版本开始去掉了对immediate参数的支持
         *                      原因是因为immediate参数会影响镜像队列的性能,增加了代码的复杂度,建议采用TTL和DLX的方法替代
         *                      当immediate参数设置为true时,如果交换器再将消息路由到一个队列中,否则将消息返回给生产者
         *                      immediate参数告诉服务器,如果该消息关联的队列上有消费者,则立刻投递
         *                      如果所有匹配的队列上都没有消费者,则直接将消息返还给生产者,不用将消息存于的队列而等待消费者
         * @param props         其它的一些属性(内容格式,类型,请求头等)
         * @param body          内容
         * @param delayTime     发送延迟时间
         * @return
         * @throws IOException
         */
        public void basicPublish(RabbitOperations channel, String exchange, String routingKey, boolean mandatory,
            boolean immediate, MessageProperties props, String body, Long delayTime) throws IOException {
            if (isBlankObject(props)) {
                props = getBasicProperties();
            }
            if (isNotBlankObject(delayTime)) {
                props.setDelay(delayTime.intValue());
            }
            // convertAndSend队列接收消息的时候是无序的
            // convertSendAndReceive队列接收消息的时候是有序的
            Message msg = new Message(body.getBytes(), props);
            channel.send(getExchangeName(exchange), getRoutingKeyName(routingKey), msg);
        }

        @Override
        public Map<String, Object> getBaseExchangeParam() {
            Map<String, Object> map = super.getBaseExchangeParam();
            // 初始化交换机参数
            initExchangeParam(map, ERabbitExchange.DIRECT);
            return map;
        }

        @Override
        public Map<String, Object> getExchangeParam() {
            Map<String, Object> map = getBaseExchangeParam();
            return map;
        }

        @Override
        public String getExchangeType() {
            return RabbitDelayMsgUtilImpl.getExchangeType();
        }

    }

    /**
     * {@link com.sojson.util.mq.rabbit.code.interfaces.impl.RabbitTopicUtilImpl}的扩展类
     * 
     * @author liu
     * @date 2022-06-17
     */
    public class RabbitTopicDelayMsgUtilImpl extends RabbitTopicUtilImpl {

        /**
         * 开始生产消息
         * 
         * @param channel
         * @param exchange      交换机的名称
         * @param routingKey    消息的Key(如果没有交换机,此时是将消息发布到的队列名)
         * @param body          内容
         * @param delayTime     发送延迟时间
         * @return
         * @throws IOException
         */
        public void basicPublish(RabbitOperations channel, String exchange, String routingKey, String body,
            Long delayTime) throws IOException {
            basicPublish(channel, exchange, routingKey, false, body, delayTime);
        }

        /**
         * 开始生产消息
         * 
         * @param channel
         * @param exchange      交换机的名称
         * @param routingKey    消息的Key(如果没有交换机,此时是将消息发布到的队列名)
         * @param mandatory     是否开启消息故障检测模式
         *                      它只会让RabbitMq向你通知失败,而不会通知成功
         *                      设为true的时候,使用前要先调用channel.addReturnListener(new ReturnCallback(){})
         *                      channel.addReturnListener(new ReturnListener(){})为异步返回
         *                      如果为true,消息不能路由到指定的队列时,会触发channel.BasicReturn事件
         *                      如果为false,则broker会直接将消息丢弃
         * @param body          内容
         * @param delayTime     发送延迟时间
         * @return
         * @throws IOException
         */
        public void basicPublish(RabbitOperations channel, String exchange, String routingKey, boolean mandatory,
            String body, Long delayTime) throws IOException {
            basicPublish(channel, exchange, routingKey, mandatory, false, getBasicProperties(), body, delayTime);
        }

        /**
         * 开始生产消息
         * 
         * @param channel
         * @param exchange      交换机的名称
         * @param routingKey    消息的Key(如果没有交换机,此时是将消息发布到的队列名)
         * @param mandatory     是否开启消息故障检测模式
         *                      它只会让RabbitMq向你通知失败,而不会通知成功
         *                      设为true的时候,使用前要先调用channel.addReturnListener(new ReturnCallback(){})
         *                      channel.addReturnListener(new ReturnListener(){})为异步返回
         *                      如果为true,消息不能路由到指定的队列时,会触发channel.BasicReturn事件
         *                      如果为false,则broker会直接将消息丢弃
         * @param immediate     RabbitMQ 3.0版本开始去掉了对immediate参数的支持
         *                      原因是因为immediate参数会影响镜像队列的性能,增加了代码的复杂度,建议采用TTL和DLX的方法替代
         *                      当immediate参数设置为true时,如果交换器再将消息路由到一个队列中,否则将消息返回给生产者
         *                      immediate参数告诉服务器,如果该消息关联的队列上有消费者,则立刻投递
         *                      如果所有匹配的队列上都没有消费者,则直接将消息返还给生产者,不用将消息存于的队列而等待消费者
         * @param props         其它的一些属性(内容格式,类型,请求头等)
         * @param body          内容
         * @param delayTime     发送延迟时间
         * @return
         * @throws IOException
         */
        public void basicPublish(RabbitOperations channel, String exchange, String routingKey, boolean mandatory,
            boolean immediate, MessageProperties props, String body, Long delayTime) throws IOException {
            if (isBlankObject(props)) {
                props = getBasicProperties();
            }
            if (isNotBlankObject(delayTime)) {
                props.setDelay(delayTime.intValue());
            }
            // convertAndSend队列接收消息的时候是无序的
            // convertSendAndReceive队列接收消息的时候是有序的
            Message msg = new Message(body.getBytes(), props);
            channel.send(getExchangeName(exchange), getRoutingKeyName(routingKey), msg);
        }

        @Override
        public Map<String, Object> getBaseExchangeParam() {
            Map<String, Object> map = super.getBaseExchangeParam();
            // 初始化交换机参数
            initExchangeParam(map, ERabbitExchange.TOPIC);
            return map;
        }

        @Override
        public Map<String, Object> getExchangeParam() {
            Map<String, Object> map = getBaseExchangeParam();
            return map;
        }

        @Override
        public String getExchangeType() {
            return RabbitDelayMsgUtilImpl.getExchangeType();
        }

    }

    /**
     * {@link com.sojson.util.mq.rabbit.code.interfaces.impl.RabbitFanoutUtilImpl}的扩展类
     * 
     * @author liu
     * @date 2022-06-17
     */
    public class RabbitFanoutDelayMsgUtilImpl extends RabbitFanoutUtilImpl {

        /**
         * 开始生产消息
         * 
         * @param channel
         * @param exchange      交换机的名称
         * @param body          内容
         * @param delayTime     发送延迟时间
         * @return
         * @throws IOException
         */
        public void basicPublish(RabbitOperations channel, String exchange, String body, Long delayTime)
            throws IOException {
            basicPublish(channel, exchange, false, false, body, delayTime);
        }

        /**
         * 开始生产消息
         * 
         * @param channel
         * @param exchange      交换机的名称
         * @param mandatory     是否开启消息故障检测模式
         *                      它只会让RabbitMq向你通知失败,而不会通知成功
         *                      设为true的时候,使用前要先调用channel.addReturnListener(new ReturnCallback(){})
         *                      channel.addReturnListener(new ReturnListener(){})为异步返回
         *                      如果为true,消息不能路由到指定的队列时,会触发channel.BasicReturn事件
         *                      如果为false,则broker会直接将消息丢弃
         * @param immediate     RabbitMQ 3.0版本开始去掉了对immediate参数的支持
         *                      原因是因为immediate参数会影响镜像队列的性能,增加了代码的复杂度,建议采用TTL和DLX的方法替代
         *                      当immediate参数设置为true时,如果交换器再将消息路由到一个队列中,否则将消息返回给生产者
         *                      immediate参数告诉服务器,如果该消息关联的队列上有消费者,则立刻投递
         *                      如果所有匹配的队列上都没有消费者,则直接将消息返还给生产者,不用将消息存于的队列而等待消费者
         * @param body          内容
         * @param delayTime     发送延迟时间
         * @return
         * @throws IOException
         */
        public void basicPublish(RabbitOperations channel, String exchange, boolean mandatory, boolean immediate,
            String body, Long delayTime) throws IOException {
            basicPublish(channel, exchange, mandatory, immediate, getBasicProperties(), body, delayTime);
        }

        /**
         * 开始生产消息
         * 
         * @param channel
         * @param exchange      交换机的名称
         * @param mandatory     是否开启消息故障检测模式
         *                      它只会让RabbitMq向你通知失败,而不会通知成功
         *                      设为true的时候,使用前要先调用channel.addReturnListener(new ReturnCallback(){})
         *                      channel.addReturnListener(new ReturnListener(){})为异步返回
         *                      如果为true,消息不能路由到指定的队列时,会触发channel.BasicReturn事件
         *                      如果为false,则broker会直接将消息丢弃
         * @param immediate     RabbitMQ 3.0版本开始去掉了对immediate参数的支持
         *                      原因是因为immediate参数会影响镜像队列的性能,增加了代码的复杂度,建议采用TTL和DLX的方法替代
         *                      当immediate参数设置为true时,如果交换器再将消息路由到一个队列中,否则将消息返回给生产者
         *                      immediate参数告诉服务器,如果该消息关联的队列上有消费者,则立刻投递
         *                      如果所有匹配的队列上都没有消费者,则直接将消息返还给生产者,不用将消息存于的队列而等待消费者
         * @param props         其它的一些属性(内容格式,类型,请求头等)
         * @param body          内容
         * @param delayTime     发送延迟时间
         * @return
         * @throws IOException
         */
        public void basicPublish(RabbitOperations channel, String exchange, boolean mandatory, boolean immediate,
            MessageProperties props, String body, Long delayTime) throws IOException {
            if (isBlankObject(props)) {
                props = getBasicProperties();
            }
            if (isNotBlankObject(delayTime)) {
                props.setDelay(delayTime.intValue());
            }
            // convertAndSend队列接收消息的时候是无序的
            // convertSendAndReceive队列接收消息的时候是有序的
            Message msg = new Message(body.getBytes(), props);
            channel.send(getExchangeName(exchange), StringUtil.EMPTY, msg);
        }

        @Override
        public Map<String, Object> getBaseExchangeParam() {
            Map<String, Object> map = super.getBaseExchangeParam();
            // 初始化交换机参数
            initExchangeParam(map, ERabbitExchange.FANOUT);
            return map;
        }

        @Override
        public Map<String, Object> getExchangeParam() {
            Map<String, Object> map = getBaseExchangeParam();
            return map;
        }

        @Override
        public String getExchangeType() {
            return RabbitDelayMsgUtilImpl.getExchangeType();
        }

    }

    /**
     * {@link com.sojson.util.mq.rabbit.code.interfaces.impl.RabbitHeadersUtilImpl}的扩展类
     * 
     * @author liu
     * @date 2022-06-17
     */
    public class RabbitHeadersDelayMsgUtilImpl extends RabbitHeadersUtilImpl {

        /**
         * 开始生产消息
         * 
         * @param channel
         * @param exchange      交换机的名称
         * @param headerParam   请求头参数
         * @param body          内容
         * @param delayTime     发送延迟时间
         * @return
         * @throws IOException
         */
        public void basicPublish(RabbitOperations channel, String exchange, String body,
            Map<String, Object> headerParam, Long delayTime) throws IOException {
            basicPublish(channel, exchange, false, false, body, headerParam, delayTime);
        }

        /**
         * 开始生产消息
         * 
         * @param channel
         * @param exchange      交换机的名称
         * @param mandatory     是否开启消息故障检测模式
         *                      它只会让RabbitMq向你通知失败,而不会通知成功
         *                      设为true的时候,使用前要先调用channel.addReturnListener(new ReturnCallback(){})
         *                      channel.addReturnListener(new ReturnListener(){})为异步返回
         *                      如果为true,消息不能路由到指定的队列时,会触发channel.BasicReturn事件
         *                      如果为false,则broker会直接将消息丢弃
         * @param immediate     RabbitMQ 3.0版本开始去掉了对immediate参数的支持
         *                      原因是因为immediate参数会影响镜像队列的性能,增加了代码的复杂度,建议采用TTL和DLX的方法替代
         *                      当immediate参数设置为true时,如果交换器再将消息路由到一个队列中,否则将消息返回给生产者
         *                      immediate参数告诉服务器,如果该消息关联的队列上有消费者,则立刻投递
         *                      如果所有匹配的队列上都没有消费者,则直接将消息返还给生产者,不用将消息存于的队列而等待消费者
         * @param body          内容
         * @param headerParam   请求头参数
         * @param delayTime     发送延迟时间
         * @return
         * @throws IOException
         */
        public void basicPublish(RabbitOperations channel, String exchange, boolean mandatory, boolean immediate,
            String body, Map<String, Object> headerParam, Long delayTime) throws IOException {
            basicPublish(channel, exchange, mandatory, immediate, getBasicProperties(), body, headerParam, delayTime);
        }

        /**
         * 开始生产消息
         * 
         * @param channel
         * @param exchange      交换机的名称
         * @param mandatory     是否开启消息故障检测模式
         *                      它只会让RabbitMq向你通知失败,而不会通知成功
         *                      设为true的时候,使用前要先调用channel.addReturnListener(new ReturnCallback(){})
         *                      channel.addReturnListener(new ReturnListener(){})为异步返回
         *                      如果为true,消息不能路由到指定的队列时,会触发channel.BasicReturn事件
         *                      如果为false,则broker会直接将消息丢弃
         * @param immediate     RabbitMQ 3.0版本开始去掉了对immediate参数的支持
         *                      原因是因为immediate参数会影响镜像队列的性能,增加了代码的复杂度,建议采用TTL和DLX的方法替代
         *                      当immediate参数设置为true时,如果交换器再将消息路由到一个队列中,否则将消息返回给生产者
         *                      immediate参数告诉服务器,如果该消息关联的队列上有消费者,则立刻投递
         *                      如果所有匹配的队列上都没有消费者,则直接将消息返还给生产者,不用将消息存于的队列而等待消费者
         * @param props         其它的一些属性(内容格式,类型,请求头等)
         * @param body          内容
         * @param headerParam   请求头参数
         * @param delayTime     发送延迟时间
         * @return
         * @throws IOException
         */
        public void basicPublish(RabbitOperations channel, String exchange, boolean mandatory, boolean immediate,
            MessageProperties props, String body, Map<String, Object> headerParam, Long delayTime) throws IOException {
            if (isBlankObject(props)) {
                props = getBasicProperties();
            }
            if (isNotBlankObject(headerParam)) {
                Set<Entry<String, Object>> entrySet = headerParam.entrySet();
                for (Entry<String, Object> entry : entrySet) {
                    props.setHeader(entry.getKey(), entry.getValue());
                }
            }
            if (isNotBlankObject(delayTime)) {
                props.setDelay(delayTime.intValue());
            }
            // convertAndSend队列接收消息的时候是无序的
            // convertSendAndReceive队列接收消息的时候是有序的
            Message msg = new Message(body.getBytes(), props);
            channel.send(getExchangeName(exchange), StringUtil.EMPTY, msg);
        }

        @Override
        public Map<String, Object> getBaseExchangeParam() {
            Map<String, Object> map = super.getBaseExchangeParam();
            // 初始化交换机参数
            initExchangeParam(map, ERabbitExchange.HEADERS);
            return map;
        }

        @Override
        public Map<String, Object> getExchangeParam() {
            Map<String, Object> map = getBaseExchangeParam();
            return map;
        }

        @Override
        public String getExchangeType() {
            return RabbitDelayMsgUtilImpl.getExchangeType();
        }

    }

    /**
     * 初始化交换机参数
     * 
     * @param param
     */
    public static void initExchangeParam(Map<String, Object> param, ERabbitExchange type) {
        param.put("x-delayed-type", type.getCode());
    }

    /**
     * 获取交换机类型
     * 
     * @return
     */
    protected static String getExchangeType() {
        return ERabbitExchange.DELAYED_MESSAGE.getCode();
    }

    /**
     * 设置消息头参数
     * 
     * @param props     其它的一些属性(内容格式,类型,请求头等)
     * @param delayTime 发送延迟时间
     * @return
     */
    protected void setConsumeHeaderParam(MessageProperties props, Long delayTime) {
        props.setHeader("x-delay", delayTime);
    }

    /**
     * 判断一个对象是否是空(空对象)
     * 
     * @param obj
     * @return
     */
    protected static boolean isBlankObject(Object obj) {
        return StringUtil.isBlankObject(obj);
    }

    /**
     * 判断一个对象是否不是空(空对象)
     * 
     * @param obj
     * @return
     */
    protected boolean isNotBlankObject(Object obj) {
        return StringUtil.isNotBlankObject(obj);
    }

    public static RabbitDelayMsgUtilImpl getInstances() {
        if (isBlankObject(instances)) {
            instances = new RabbitDelayMsgUtilImpl();
        }
        return instances;
    }

}